package com.iflytek.uoamp.agent.monitor.sender.schedual;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.log4j.Logger;

import com.iflytek.uoamp.agent.common.util.LogHelper;
import com.iflytek.uoamp.agent.common.util.ReadFileByLine;
import com.iflytek.uoamp.agent.common.util.ScanFile;
import com.iflytek.uoamp.agent.common.util.WFile;
import com.iflytek.uoamp.agent.core.sender.monitorsender.monitorinterface.IItem;
import com.iflytek.uoamp.agent.monitor.mcagent.SendClient;
import com.iflytek.uoamp.agent.monitor.sender.biz.AbsSender;
import com.iflytek.uoamp.agent.monitor.sender.config.Config;
import com.iflytek.uoamp.agent.monitor.sender.interfaces.ISender;

public class ScheduleRunning implements Runnable {

	private static final int READLINE = 10;

	private Config config = null;
	private ExecutorService callTimeoutPool;
	private final AtomicLong threadCounter;

	private String cachePath = null;
	private AbsSender sender;
	private ScanFile scaner = new ScanFile();
	long starttime;
	private ISender client = SendClient.getInstance();
	private void parseFile(String file, int line) {
		Logger.getRootLogger().info("parse file " + file + " and " + line);
		List<String> list = ReadFileByLine.getAllLine2Array(file, "utf-8", line, READLINE);
		int skipLine = line;
		while (list != null && list.size() != 0) {
			if ((System.currentTimeMillis() - starttime) > 10 * 1000) {
				return;
			}
			for (String cur : list) {
				skipLine++;
				IItem item = sender.getFactory().caseObject(cur);
				send(item);
				WriteFileIndex(file, skipLine);
			}
			list = ReadFileByLine.getAllLine2Array(file, "utf-8", skipLine, READLINE);
		}

		new File(file).delete();
		Logger.getRootLogger().info("delete cache file " + file);
		WriteFileIndex("", 0);
	}

	@SuppressWarnings("unused")
	private void cleanEmptyFile(String file) {
		ScanFile scaner = new ScanFile();
		ArrayList<String> files = scaner.seanFile(file);
		for (String cur : files) {
			if (!cur.contains("no_send.")) {
				continue;
			}
			File curFile = new File(cur);
			if (curFile.length() == 0) {
				curFile.delete();
			}
		}
	}

	private void send(IItem iItem) {
//		ConnectionPoolManager manager = null;
//		ISender client = null;
		try {
//			manager = sender.getLbManager().getNextConnectManager();
//			if (manager == null) {
//				LogHelper.error(Config.getCache(), iItem.caseToString() + "\n");
//			} else {
//				client = manager.checkout();
				if (client != null) {
					try{
						doAppend(client, iItem).get(config.getRequestTimeOut(), TimeUnit.MILLISECONDS);
					}catch(Exception e){
						Logger.getRootLogger().error(e.getMessage(),e);
						if (iItem != null) {
							LogHelper.error(Config.getCache(), iItem.caseToString() + "\n");
						}
					}
//					manager.checkIn(client);
					LogHelper.info(Config.getLog(), "send item by ScheduleRunning...");
				} else {
					LogHelper.error(Config.getCache(), iItem.caseToString() + "\n");
				}

//			}
		} catch (Exception e) {
			Logger.getRootLogger().error(e.getMessage(),e);
			LogHelper.error(Config.getLog(),ExceptionUtils.getStackTrace(e));
		}
	}

	private Future<Object> doAppend(final ISender client, final IItem e) throws Exception {
		return callTimeoutPool.submit(new Callable<Object>() {
			@Override
			public Object call() throws Exception {
				Object result = client.send(e);
				return result;
			}
		});
	}

	// 读取转移了一半的文件信息
	private int getFileIndex(StringBuffer filename) {
		if (new File(cachePath + "/file.index").exists() == false) {
			WFile.wf(cachePath + "/file.index", "0", false);
		}

		List<String> lines = ReadFileByLine.getAllLine2Array(cachePath + "/file.index");
		if (lines.size() != 2) {
			return 0;
		} else {
			filename.append(lines.get(0));
			Logger.getRootLogger().info("read " + lines.get(0) + "  " + lines.get(1));
			return Integer.parseInt(lines.get(1));
		}
	}


	// 写入本次转移的位置到file.index
	private void WriteFileIndex(String filename, int index) {
		WFile.wf(cachePath + "/file.index", filename, false);
		WFile.wf(cachePath + "/file.index", "\n" + index, true);
	}


	public ScheduleRunning(AbsSender sender) {
		this.config = sender.getConfig();
		this.sender = sender;
		threadCounter = new AtomicLong(0);
		cachePath = this.config.getCachePath();
		callTimeoutPool = Executors.newCachedThreadPool(new ThreadFactory() {
			@Override
			public Thread newThread(Runnable r) {
				Thread t = new Thread(r);
				t.setName("Flume Thrift ScheNoSend thread - " + String.valueOf(threadCounter.incrementAndGet()));
				return t;
			}
		});
	}

	@Override
	public void run() {
		try {
			// cleanEmptyFile(cachePath);
			starttime = System.currentTimeMillis();

			ArrayList<String> files = scaner.seanFile(cachePath);
			Logger.getRootLogger().debug("DEBUG | start parse cache file!! " + files.size() + " sacan " + cachePath);
			StringBuffer inTransferFile = new StringBuffer();
			int FileReadLine = getFileIndex(inTransferFile);
			if (FileReadLine != 0) {
				parseFile(inTransferFile.toString(), FileReadLine);
			}

			for (String currentFilename : files) {

				if ((System.currentTimeMillis() - starttime) > 10 * 1000) {
					return;
				}

				if (inTransferFile.toString().equals(currentFilename)) {
					continue;
				}

				if (!currentFilename.contains("no_send.")) {
					continue;
				}

				parseFile(currentFilename, 0);
				LogHelper.info(Config.getLog(), "filenames:" + currentFilename);

			}
		} catch (Exception e) {
			Logger.getRootLogger().error(e.getMessage(),e);
			LogHelper.error(Config.getLog(),ExceptionUtils.getStackTrace(e));
		}
	}
}
